////
// @file block.cc
// @brief
// 实现block
//
// @author niexw
// @email niexiaowen@uestc.edu.cn
//
#include <algorithm>
#include <cmath>
#include <db/block.h>
#include <db/record.h>
#include <db/table.h>

namespace db {
DataBlock::RecordIterator::RecordIterator()
    : block(nullptr)
    , index(0)
{}
DataBlock::RecordIterator::~RecordIterator() {}
DataBlock::RecordIterator::RecordIterator(const RecordIterator &other)
    : block(other.block)
    , record(other.record)
    , index(other.index)
{}

DataBlock::RecordIterator &DataBlock::RecordIterator::operator++()
{
    if (block == nullptr || block->getSlots() == 0) return *this;
    index = (++index) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return *this;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return *this;
}
DataBlock::RecordIterator DataBlock::RecordIterator::operator++(int)
{
    RecordIterator tmp(*this);
    if (block == nullptr || block->getSlots() == 0) return tmp;
    index = (++index) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return tmp;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return tmp;
}
DataBlock::RecordIterator &DataBlock::RecordIterator::operator--()
{
    if (block == nullptr || block->getSlots() == 0) return *this;
    index = (index + block->getSlots()) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return *this;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return *this;
}
DataBlock::RecordIterator DataBlock::RecordIterator::operator--(int)
{
    RecordIterator tmp(*this);
    if (block == nullptr || block->getSlots() == 0) return tmp;
    index = (index + block->getSlots()) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return tmp;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return tmp;
}
Record *DataBlock::RecordIterator::operator->() { return &record; }
DataBlock::RecordIterator &DataBlock::RecordIterator::operator+=(int step)
{
    if (block == nullptr || block->getSlots() == 0) return *this;
    index = (index + step) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return *this;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return *this;
}
DataBlock::RecordIterator &DataBlock::RecordIterator::operator-=(int step)
{
    if (block == nullptr || block->getSlots() == 0) return *this;
    index = (index + step + block->getSlots()) % (block->getSlots() + 1);
    if (index == block->getSlots()) {
        record.detach();
        return *this;
    }
    Slot *slots = block->getSlotsPointer();
    record.attach(
        block->buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    return *this;
}

void SuperBlock::clear(const unsigned short spaceId) const
{
    // 清buffer
    memset(buffer_, 0, SUPER_SIZE);
    auto header = reinterpret_cast<SuperHeader *>(buffer_);

    // 设置magic number
    header->magic = MAGIC_NUMBER;
    // 设定space id
    setSpaceId(spaceId);
    // 设定类型
    setType(BLOCK_TYPE_SUPER);
    // 设定时戳
    setTimeStamp();
    // 设定数据块
    setFirst(0);
    // 设定max id
    setMaxId(0);
    // 设定self
    setSelf();
    // 设定空闲块，缺省从1开始
    setIdle(0);
    // 设定记录数目
    setRecords(0);
    // 设定数据块个数
    setDataCounts(0);
    // 设定空闲块个数
    setIdleCounts(0);
    // 设定空闲空间
    setFreeSpace(sizeof(SuperHeader));
    // 设置checksum
    setChecksum();
}

void MetaBlock::clear(
    const unsigned short spaceId,
    const unsigned int self,
    const unsigned short type) const
{
    // 清buffer
    memset(buffer_, 0, BLOCK_SIZE);
    auto header = reinterpret_cast<MetaHeader *>(buffer_);

    // 设定magic
    header->magic = MAGIC_NUMBER;
    // 设定space id
    setSpaceId(spaceId);
    // 设定类型
    setType(type);
    // 设定空闲块
    setNext(0);
    // 设置本块id
    setSelf(self);
    // 设定时戳
    setTimeStamp();
    // 设定slots
    setSlots(0);
    // 设定free size
    setFreeSize(BLOCK_SIZE - sizeof(MetaHeader) - sizeof(Trailer));
    // 设定free space
    setFreeSpace(sizeof(MetaHeader));
    // 设定校验和
    setChecksum();
}

// TODO: 如果record非full，直接分配，不考虑slot
std::pair<unsigned char *, bool>
MetaBlock::allocate(unsigned short space, unsigned short index) const
{
    bool need_reorder = false;
    space = ALIGN_TO_SIZE(space); // 先将需要空间数对齐8B

    // 计算需要分配的空间，需要考虑到分配Slot的问题
    unsigned short demand_space = space;
    unsigned short freeSize = getFreeSize(); // block当前的剩余空间
    unsigned short currentTrailerSize = getTrailerSize();
    unsigned short demandTrailerSize =
        (getSlots() + 1) * sizeof(Slot) + sizeof(int);
    if (currentTrailerSize < demandTrailerSize)
        demand_space += ALIGN_TO_SIZE(sizeof(Slot)); // 需要的空间数目

    // 该block空间不够
    if (freeSize < demand_space)
        return std::pair<unsigned char *, bool>(nullptr, false);

    // 如果free space空间不够，先回收删除的记录
    unsigned short freeSpaceSize = getFreeSpaceSize();
    // free space的空间要减去要分配的slot的空间
    if (currentTrailerSize < demandTrailerSize)
        freeSpaceSize -= ALIGN_TO_SIZE(sizeof(Slot));
    // NOTE: 这里这里没法reorder，才分配还未填充记录
    if (freeSpaceSize < demand_space) {
        shrink();
        need_reorder = true;
    }

    // 从free space分配空间
    unsigned char *ret = buffer_ + getFreeSpace();

    // 增加slots计数
    unsigned short old = getSlots();
    unsigned short total = std::min<unsigned short>(old, index);
    // READING: slots[]放在block的结尾，提前设置slots的值相当于在原slots[]开头前开辟了一块空间
    setSlots(old + 1);
    // 在slots[]顶部增加一个条目
    Slot *new_position = getSlotsPointer();
    for (unsigned short i = 0; i < total; ++i, ++new_position)
        *new_position = *(new_position + 1);
    new_position = getSlotsPointer() + index;
    // READING: 指定新分配的槽位
    new_position->offset = htobe16(getFreeSpace());
    new_position->length = htobe16(space);

    // 设定空闲空间大小
    setFreeSize(getFreeSize() - demand_space);
    // 设定free space偏移量
    // READING: free space 相当于块内指针
    setFreeSpace(getFreeSpace() + space);

    return std::pair<unsigned char *, bool>(ret, need_reorder);
}

// TODO: 需要考虑record非full的情况
void MetaBlock::deallocate(const unsigned short index) const
{
    // 计算需要删除的记录的槽位
    auto pSlot = reinterpret_cast<Slot *>(
        buffer_ + BLOCK_SIZE - sizeof(int) -
        sizeof(Slot) * (getSlots() - index));
    Slot slot;
    slot.offset = be16toh(pSlot->offset);
    slot.length = be16toh(pSlot->length);

    // 设置tombstone
    Record record;
    unsigned char *space = buffer_ + slot.offset;
    record.attach(space, 8); // 只使用8个字节
    // READING: Header 1B, Length 1-4B, Aligned Length 8B, 8B是对齐的最小长度了,已经包含所有有效信息
    // READING: 移动Record代价太大，不显式移动
    record.die();

    // 挤压slots[]
    // READING: 由于slots[]从开头分配/回收，所以需要将前面的slots[]向后挤压
    for (unsigned short i = index; i > 0; --i) {
        Slot *from = pSlot;
        --from;
        pSlot->offset = from->offset;
        pSlot->length = from->length;
        pSlot = from;
    }

    // 回收slots[]空间
    unsigned short previousTrailerSize = getTrailerSize();
    // READING: 相当于回收slots[0]的空间
    setSlots(getSlots() - 1);
    unsigned short currentTrailerSize = getTrailerSize();
    // 要把slots[]回收的空间加回来
    // READING: 主要是为了对齐
    if (previousTrailerSize > currentTrailerSize)
        slot.length += previousTrailerSize - currentTrailerSize;
    // 修改free size
    setFreeSize(getFreeSize() + slot.length);
}

void MetaBlock::shrink() const
{
    Slot *slots = getSlotsPointer();

    // 按照偏移量重新排序slots[]函数
    struct OffsetSort
    {
        bool operator()(const Slot &x, const Slot &y) const
        {
            return be16toh(x.offset) < be16toh(y.offset);
        }
    };
    OffsetSort oSort;
    std::sort(slots, slots + getSlots(), oSort);

    // 枚举所有record，然后向前移动
    unsigned short offset = sizeof(MetaHeader);
    unsigned short space = 0;
    for (unsigned short i = 0; i < getSlots(); ++i) {
        unsigned short len = be16toh((slots + i)->length);
        unsigned short off = be16toh((slots + i)->offset);
        if (offset < off) memmove(buffer_ + offset, buffer_ + off, len);
        (slots + i)->offset = htobe16(offset);
        offset += len;
        space += len;
    }

    // 设定free space
    setFreeSpace(offset);
    // 计算free size
    setFreeSize(BLOCK_SIZE - sizeof(MetaHeader) - getTrailerSize() - space);
}

unsigned short DataBlock::searchRecord(void *buf, size_t len) const
{
    // 获取key位置
    RelationInfo *info = table_->info_;
    unsigned int key = info->key;

    // 调用数据类型的搜索
    return info->fields[key].type->search(buffer_, key, buf, len);
}

std::pair<unsigned short, bool>
DataBlock::splitPosition(size_t space, unsigned short index) const
{
    static constexpr unsigned short BlockHalf =
        (BLOCK_SIZE - sizeof(DataHeader) - sizeof(unsigned int)) / 2; // 一半的大小

    // 枚举所有记录
    unsigned short count = getSlots();
    size_t half = 0;
    Slot *slots = getSlotsPointer();
    bool included = false;
    unsigned short i;
    for (i = 0; i < count; ++i) {
        // 如果是index，则将需要插入的记录空间算在内
        if (i == index) {
            // 这里的计算并不精确，没有准确考虑slot的大小，但只算一半没有太大的误差。
            half += ALIGN_TO_SIZE(space) + sizeof(Slot);
            if (half > BlockHalf)
                break;
            included = true;
        }

        // fallthrough, i != index
        half += be16toh(slots[i].length);
        if (half > BlockHalf) break;
    }
    return std::pair<unsigned short, bool>(i, included);
}

unsigned short DataBlock::requireLength(const std::vector<struct iovec> &iov) const
{
    size_t length = ALIGN_TO_SIZE(Record::size(iov)); // 对齐8B后的长度
    size_t trailer =
        ALIGN_TO_SIZE((getSlots() + 1) * sizeof(Slot) + sizeof(unsigned int)) -
        ALIGN_TO_SIZE(
            getSlots() * sizeof(Slot) +
            sizeof(unsigned int)); // trailer新增部分
    return static_cast<unsigned short>(length + trailer);
}

std::pair<bool, unsigned short>
DataBlock::insertRecord(const std::vector<struct iovec> &iov) const
{
    RelationInfo *info = table_->info_;
    unsigned int key = info->key;
    DataType *type = info->fields[key].type;

    // 比较key
    Record record;
    auto refResult = refRecord(record, iov[key].iov_base, static_cast<unsigned int>(iov[key].iov_len));
    auto index = refResult.second;
    auto result = refResult.first;
    if (result)
        return std::make_pair(false, -1);

    // 如果block空间足够，插入
    size_t bLen = getFreeSize(); // 该block的富余空间
    auto actLen = static_cast<unsigned short>(Record::size(iov));
    // unsigned short alignLen = ALIGN_TO_SIZE(actLen);
    unsigned short trailerLen =
        ALIGN_TO_SIZE((getSlots() + 1) * sizeof(Slot) + sizeof(unsigned int)) -
        ALIGN_TO_SIZE(getSlots() * sizeof(Slot) + sizeof(unsigned int));
    if (bLen < actLen + trailerLen)
        return std::make_pair(false, index);

    // 分配空间
    std::pair<unsigned char *, bool> alloc_ret = allocate(actLen, index);
    // 填写记录
    record.attach(alloc_ret.first, actLen);
    unsigned char header = 0;
    record.set(iov, &header);
    // 重新排序
    if (alloc_ret.second) reorder(type, key);

    return std::make_pair(true, index);
}

std::pair<bool, unsigned short> DataBlock::refRecord(Record &record, void *keyBuf, unsigned int len) const
{
    RelationInfo *info = table_->info_;
    unsigned int key = info->key;
    DataType *type = info->fields[key].type;

    // 先确定插入位置
    unsigned short index =
        type->search(buffer_, key, keyBuf, len);
    if (index >= getSlots()) {
        record.detach();
        return std::make_pair(false, getSlots()); // 没有找到记录
    }
    Slot *slots = getSlotsPointer();
    record.attach(
        buffer_ + be16toh(slots[index].offset),
        be16toh(slots[index].length));
    unsigned char *pkey;
    unsigned int pLen;
    record.refByIndex(&pkey, &pLen, key);
    if (memcmp(pkey, keyBuf, pLen) != 0) {
        record.detach(); // key不匹配
        return std::make_pair(false, index);
    }
    // key匹配，返回记录
    return std::make_pair(true, index);
}

bool DataBlock::updateRecord(const std::vector<struct iovec> &iov) const
{
    RelationInfo *info = table_->info_;
    if (info == nullptr) return false;
    unsigned int key = info->key;

    // 比较key
    Record record;
    auto refResult = refRecord(record, iov[key].iov_base, static_cast<unsigned int>(iov[key].iov_len));
    auto index = refResult.second;
    auto result = refResult.first;
    if (!result) {
        return false; // 没有找到记录
    }

    // TODO: 是否重排后再插入?
    deallocate(index); // 先删除记录
    // 再插入
    std::pair<bool, unsigned short> ret = insertRecord(iov);
    if (ret.first) {
        return true; // 插入成功
    }
    // 插入失败，恢复原记录
    std::vector<struct iovec> old_iov;
    unsigned char header;
    record.ref(old_iov, &header); // 重新引用原记录
    record.detach();
    shrink();
    // TODO: 表达式未使用
    insertRecord(old_iov); // 恢复原记录
    return false; // 插入失败
}

bool DataBlock::removeRecord(void *keyBuf, unsigned int len) const
{
    Record record;
    auto refResult = refRecord(record, keyBuf, len);
    auto index = refResult.second;
    auto result = refResult.first;
    if (!result) {
        return false; // 没有找到记录
    }
    deallocate(index); // 删除记录
    // TODO: 检查shrink作用
    // shrink(); // 回收资源
    return true; // 删除成功
}

bool DataBlock::copyRecord(const Record &record) const
{
    // 判断剩余空间是否足够
    size_t bLen = getFreeSpaceSize(); // 该block的富余空间
    unsigned short actLen = static_cast<unsigned short>(record.allocLength());
    unsigned short trailerLen =
        ALIGN_TO_SIZE((getSlots() + 1) * sizeof(Slot) + sizeof(unsigned int)) -
        ALIGN_TO_SIZE(getSlots() * sizeof(Slot) + sizeof(unsigned int));
    if (bLen < actLen + trailerLen) return false;

    // 分配空间，然后copy
    std::pair<unsigned char *, bool> alloc_ret = allocate(actLen, getSlots());
    memcpy(alloc_ret.first, record.buffer_, actLen);

    // TODO: 最后重排
#if 0
    // 重新排序，最后才重拍？
    RelationInfo *info = table_->info_;
    unsigned int key = info->key;
    DataType *type = info->fields[key].type;
    reorder(type, key); // 最后才重排？
#endif
    return true;
}

DataBlock::RecordIterator DataBlock::beginRecord()
{
    RecordIterator ri;
    ri.block = this;
    ri.index = 0;

    if (getSlots()) {
        Slot *slots = getSlotsPointer();
        ri.record.attach(
            buffer_ + be16toh(slots[0].offset), be16toh(slots[0].length));
    }
    return ri;
}
DataBlock::RecordIterator DataBlock::endRecord()
{
    RecordIterator ri;
    ri.block = this;
    ri.index = getSlots();
    return ri;
}

void BTreeBlock::clear(
    const unsigned short spaceId,
    const unsigned int self)
{
    memset(buffer_, 0, BLOCK_SIZE);
    auto header = reinterpret_cast<BTreeHeader *>(buffer_);

    header->magic = MAGIC_NUMBER;
    setSpaceId(spaceId);
    setType(BLOCK_TYPE_BTREE_INNER);
    setSelf(self);
    setTimeStamp();
    setSize(0);
    setFreeSize(BLOCK_SIZE - sizeof(BTreeHeader) - sizeof(BTreeTrailer));
    setFreeSpace(sizeof(BTreeHeader));
    setChecksum();
}

unsigned int BTreeBlock::searchNode(void * keyBuf, size_t keyLen) const
{
    if (buffer_ == nullptr || keyBuf == nullptr || keyLen == 0) return -1;
    BTreeNode *nodes = getNodesPointer();
    if (nodes == nullptr) return -1;
    unsigned int i = 0;
    // struct iovec iov;
    void *iovPtr;
    unsigned int iovLen;
    refKey(0, &iovPtr, &iovLen);
    DataType *keyType = table_->info_->fields[table_->info_->key].type;
    while (i < getSize() && keyType->less(
                                reinterpret_cast<unsigned char *>(keyBuf),
                                static_cast<unsigned int>(keyLen),
                                reinterpret_cast<unsigned char *>(iovPtr),
                                iovLen)) {
        ++i;
        refKey(i, &iovPtr, &iovLen);
    }
    return i; // 返回第一个大于等于key的节点
}

unsigned char * BTreeBlock::allocate(unsigned short space, unsigned short index, unsigned int data) const
{
    if (buffer_ == nullptr)return nullptr;
    space = ALIGN_TO_SIZE(space);

    unsigned short demand_space = space;
    unsigned short currentTrailerSize = getTrailerSize();
    unsigned short demandTrailerSize =
        (getSize() + 1) * sizeof(BTreeNode) + sizeof(BTreeTrailer);
    if (currentTrailerSize < demandTrailerSize)
        demand_space += ALIGN_TO_SIZE(sizeof(BTreeNode)); // 需要的空间数

    if (getFreeSize() < demand_space)
        return nullptr;

    unsigned short freeSpaceSize = getFreeSpaceSize();
    if (currentTrailerSize < demandTrailerSize)
        freeSpaceSize -= ALIGN_TO_SIZE(sizeof(BTreeNode));
    if (freeSpaceSize < demand_space) {
        shrink();
    }
    // 从free space分配空间
    unsigned char *ret = buffer_ + getFreeSpace();
    // 增加size
    unsigned short old = getSize();
    unsigned short total = std::min<unsigned short>(old, index);
    setSize(old + 1);
    BTreeNode *new_position = getNodesPointer();
    for (unsigned short i = 0; i < total; ++i, ++new_position)
        *new_position = *(new_position + 1);
    new_position = getNodesPointer() + index;
    new_position->key.offset = htobe16(getFreeSpace());
    new_position->key.length = htobe16(space);
    std::swap(new_position->data, (new_position + 1)->data);
    (new_position + 1)->data = htobe32(data);

    setFreeSize(getFreeSize() - demand_space);
    setFreeSpace(getFreeSpace() + space);
    return ret;
}

void BTreeBlock::shrink() const
{
    auto nodes = getNodesPointer();
    unsigned short offset = sizeof(BTreeHeader);
    unsigned short space = 0;
    for (unsigned short i = 0; i < getSize(); ++i) {
        unsigned short len = be16toh((nodes + i)->key.length);
        unsigned short off = be16toh((nodes + i)->key.offset);
        if (offset < off) memmove(buffer_ + offset, buffer_ + off, len);
        (nodes + i)->key.offset = htobe16(offset);
        offset += len;
        space += len;
    }

    setFreeSpace(offset);
    setFreeSize(BLOCK_SIZE - sizeof(BTreeHeader) - getTrailerSize() - space);
}

void BTreeBlock::deallocate(const unsigned short index) const
{
    if (buffer_ == nullptr || index >= getSize()) return;
    BTreeNode *pNode = getNodesPointer() + (getSize() - index);
    Slot slot;
    slot.offset = be16toh(pNode->key.offset);
    slot.length = be16toh(pNode->key.length);
    (pNode+1)->data = pNode->data;

    for (unsigned short i = 0; i < getSize(); --i) {
        BTreeNode *from = pNode;
        --from;
        pNode->key = from->key;
        (pNode+1)->data = (from+1)->data;
        pNode = from;
    }

    unsigned short previousTrailerSize = getTrailerSize();
    setSize(getSize() - 1);
    unsigned short currentTrailerSize = getTrailerSize();

    if (previousTrailerSize > currentTrailerSize)
        slot.length += previousTrailerSize - currentTrailerSize;
    setFreeSize(getFreeSize() + slot.length);
}


} // namespace db
